/**
 * 
 */
package hbec.kettle.etl.plugins.core;

import java.util.Map;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.job.Job;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import hbec.kettle.etl.common.exception.ExceptionTracker;
import hbec.kettle.etl.common.plugin.JobStartable;
import hbec.kettle.etl.common.plugin.Result;

/**
 * @author Administrator
 *
 */
public class JobStarter implements JobStartable {
	private static final Logger logger = LoggerFactory.getLogger(JobStarter.class);
	
	private Job job;
	private LogLevel logLevel;
	
	@Override
	public void init() {
		try {
			PluginEnvironment.init();
		} catch (KettleException e) {
			logger.error(ExceptionTracker.trace(e));
		}
	}

	@Override
	public void setJob(Job job) {
		this.job = job;
	}

	@Override
	public void setVariables(Map<String, String> variables) {
		for (Map.Entry<String, String> var : variables.entrySet()) {
			job.setVariable(var.getKey(), var.getValue());
		}
	}

	@Override
	public void setLogLevel(LogLevel logLevel) {
		this.logLevel = logLevel;
	}
	
	@Override
	public Result start() {
		Result res = new Result();
		
		// 设置Job日志等级
        job.setLogLevel(logLevel);
        // 开始执行Job
        job.start();
        // 等待Job执行结束 
        job.waitUntilFinished();
        // 处理作业反馈信息
        submitResult(job, res);
        // 完成
        job.setFinished(true);
        
		return res;
	}

	private static void submitResult(Job job, Result res) {
		res.result = job.getResult().getResult();
		
		res.entryNr = job.getResult().getEntryNr();
		res.nrErrors = job.getResult().getNrErrors();
		res.nrLinesInput = job.getResult().getNrLinesInput();
		res.nrLinesOutput = job.getResult().getNrLinesOutput();
		res.nrLinesUpdated = job.getResult().getNrLinesUpdated();
		res.nrLinesRead = job.getResult().getNrLinesRead();
		res.nrLinesWritten = job.getResult().getNrLinesWritten();
		res.nrLinesDeleted = job.getResult().getNrLinesDeleted();
		res.nrFilesRetrieved = job.getResult().getNrFilesRetrieved();
		res.nrLinesRejected = job.getResult().getNrLinesRejected();
		
		res.logChannelId = job.getResult().getLogChannelId();
		res.logText = job.getResult().getLogText();
		
		res.stopped = job.getResult().stopped;
		res.exitStatus = job.getResult().getExitStatus();
	}
}
